package com.xl.competition.old.readMysql2Hive

/**
 * Author: xl
 * Date: 2022/4/12 20:21
 * program: com.xl.competition
 * Email: 2199396150@qq.com
 * Description: ${description}
 */

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.sql.functions.{col, substring}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object test {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    System.setProperty("HADOOP_USER_NAME", "xxl")
    val sparkconf: SparkConf = new SparkConf().setAppName("write")
      .setMaster("local[*]")
    val spark = SparkSession.builder()
      .enableHiveSupport()
      .config(sparkconf)
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .getOrCreate()
    val customer: DataFrame = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://competition1:3306/competition")
      .option("user", "root")
      .option("password", "Abc123..")
      .option("dbtable", "customer")
      .load()
    spark.sql("use ods")
    spark.sql(
      """
        |create table if not exists customer (
        |custkey int,
        |Name string,
        |gender string,
        |address string,
        |nationkey int,
        |phone string,
        |acctbal string,
        |mktsegment string,
        |times string,
        |datimes string
        |) partitioned by (date string)
        |row format delimited
        |fields terminated by "\t"
        |""".stripMargin)
    val value: Dataset[Row] = customer
      .select(
        substring(col("datime"), 0, 10)
      ).distinct()

    val rows: Array[String] = value.rdd.map(
      (lien: Row) => {
        val str = lien.getString(0)
        str
      }
    ).collect()
    customer.createTempView("MysqlCustomer")
    rows.foreach(
      (lien: String) => {
        println(lien)
        spark.sql(
          s"""
             |INSERT overwrite table customer partition(date="${lien}")
             |select * from MysqlCustomer where substring(datime,1,10)="${lien}"
             |""".stripMargin)
      }
    )
    spark.stop()
  }
}

